package drds.data_propagate.binlog_event;

import drds.data_propagate.binlog_event.binlog_event.*;
import drds.data_propagate.binlog_event.binlog_event.binlog_management.*;
import drds.data_propagate.binlog_event.binlog_event.load_infile_replication.*;
import drds.data_propagate.binlog_event.binlog_event.mariadb.*;
import drds.data_propagate.binlog_event.binlog_event.row_based_replication_events.*;
import drds.data_propagate.binlog_event.binlog_event.statement_based_replication_events.*;
import drds.data_propagate.driver.packets.GtidSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.BitSet;


public final class BinLogEventDecoder {

    protected static final Logger log = LoggerFactory.getLogger(BinLogEventDecoder.class);

    protected final BitSet handleEventTypeSet = new BitSet(BinlogEventType.binlog_event_enum_end_event);

    public BinLogEventDecoder() {
    }

    public BinLogEventDecoder(final int fromIndex, final int toIndex) {
        handleEventTypeSet.set(fromIndex, toIndex);
    }

    public BinLogEvent decode(Buffer buffer, Header header, BinLogEventsContext binLogEventsContext) throws IOException {
        FormatDescriptionEvent formatDescriptionEvent = binLogEventsContext.getFormatDescriptionEvent();
        BinlogEventPosition binlogEventPosition = binLogEventsContext.getBinlogEventPosition();

        int checkSumAlg = BinLogEvent.binlog_check_sum_alg_undef;
        if (header.getEventType() != BinlogEventType.binlog_event_format_description_event) {
            checkSumAlg = formatDescriptionEvent.header.getChecksumAlg();
        } else {
            // 如果是format事件自己，也需要处理checksum
            checkSumAlg = header.getChecksumAlg();
        }

        if (checkSumAlg != BinLogEvent.binlog_checksum_alg_off
                && checkSumAlg != BinLogEvent.binlog_check_sum_alg_undef) {
            // remove checksum bytes
            buffer.newLimit(header.getEventLength() - BinLogEvent.binlog_checksum_len);
        }

        GtidEvent gtidEvent = binLogEventsContext.getGtidEvent();
        // GtidSet gtidSet = binLogEventsContext.getGtidSet();
        String binlogEventType = BinlogEventType.getTypeName(header.getEventType());
        log.info("binlogEventType:" + binlogEventType + ",logFileName:" + header.getLogFileName() + ",nextBinLogEventPosition:" + header.getNextBinLogEventPosition());
        switch (header.getEventType()) {
            case BinlogEventType.binlog_event_start_event_v3: {
                /* This is sent only by MySQL <=4.x */
                StartEventV3 startEventV3 = new StartEventV3(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return startEventV3;
            }
            case BinlogEventType.binlog_event_query_event: {
                QueryEvent queryEvent = new QueryEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return queryEvent;
            }
            case BinlogEventType.binlog_event_stop_event: {
                StopEvent stopEvent = new StopEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return stopEvent;
            }
            case BinlogEventType.binlog_event_rotate_event: {
                RotateEvent rotateEvent = new RotateEvent(header, buffer, formatDescriptionEvent);
                //在这里进行设置
                /* updating readedIndex in context */
                binlogEventPosition = new BinlogEventPosition(rotateEvent.getFileName(), rotateEvent.getPosition());
                binLogEventsContext.setBinlogEventPosition(binlogEventPosition);
                return rotateEvent;
            }
            case BinlogEventType.binlog_event_intvar_event: {
                IntvarEvent intvarEvent = new IntvarEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return intvarEvent;
            }
            //

            case BinlogEventType.binlog_event_load_event:
            case BinlogEventType.binlog_event_new_load_event: {
                LoadEvent loadEvent = new LoadEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return loadEvent;
            }
            case BinlogEventType.binlog_event_slave_event: /* can never happen (unused binlog_event) */ {
                if (log.isWarnEnabled())
                    log.warn("Skipping unsupported SLAVE_EVENT from: " + binLogEventsContext.getBinlogEventPosition());
                break;
            }
            case BinlogEventType.binlog_event_create_file_event: {
                CreateFileEvent createFileEvent = new CreateFileEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return createFileEvent;
            }
            case BinlogEventType.binlog_event_append_block_event: {
                AppendBlockEvent appendBlockEvent = new AppendBlockEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return appendBlockEvent;
            }
            case BinlogEventType.binlog_event_exec_load_event: {
                ExecLoadEvent execLoadEvent = new ExecLoadEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return execLoadEvent;
            }
            //
            case BinlogEventType.binlog_event_delete_file_event: {
                DeleteFileEvent deleteFileEvent = new DeleteFileEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return deleteFileEvent;
            }
            //case BinlogEventType.binlog_event_new_load_event: {上面
            case BinlogEventType.binlog_event_rand_event: {
                RandEvent randEvent = new RandEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return randEvent;
            }
            case BinlogEventType.binlog_event_user_var_event: {
                UserVarEvent userVarEvent = new UserVarEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return userVarEvent;
            }
            case BinlogEventType.binlog_event_format_description_event: {
                formatDescriptionEvent = new FormatDescriptionEvent(header, buffer, formatDescriptionEvent);
                binLogEventsContext.setFormatDescriptionEvent(formatDescriptionEvent);
                return formatDescriptionEvent;
            }
            //


            case BinlogEventType.binlog_event_xid_event: {
                XidEvent xidEvent = new XidEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return xidEvent;
            }
            case BinlogEventType.binlog_event_begin_load_query_event: {
                BeginLoadQueryEvent beginLoadQueryEvent = new BeginLoadQueryEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return beginLoadQueryEvent;
            }
            case BinlogEventType.binlog_event_execute_load_query_event: {
                ExecuteLoadQueryEvent executeLoadQueryEvent = new ExecuteLoadQueryEvent(header, buffer,
                        formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return executeLoadQueryEvent;
            }
            case BinlogEventType.binlog_event_table_map_event: {
                TableMapEvent tableMapEvent = new TableMapEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                binLogEventsContext.getTableIdToTableMapEventMap().put(tableMapEvent.getTableId(), tableMapEvent);
                return tableMapEvent;
            }

            case BinlogEventType.binlog_event_write_rows_event_v0: {
                if (log.isWarnEnabled())
                    log.warn("Skipping unsupported PRE_GA_WRITE_ROWS_EVENT from: " + binLogEventsContext.getBinlogEventPosition());
                // ev = new Write_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            }
            //
            case BinlogEventType.binlog_event_update_rows_event_v0: {
                if (log.isWarnEnabled())
                    log.warn("Skipping unsupported PRE_GA_UPDATE_ROWS_EVENT from: " + binLogEventsContext.getBinlogEventPosition());
                // ev = new Update_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            }
            case BinlogEventType.binlog_event_delete_rows_event_v0: {
                if (log.isWarnEnabled())
                    log.warn("Skipping unsupported PRE_GA_DELETE_ROWS_EVENT from: " + binLogEventsContext.getBinlogEventPosition());
                // ev = new Delete_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            }
            case BinlogEventType.binlog_event_write_rows_event_v1: {
                RowsEvent rowsEvent = new WriteRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                rowsEvent.setTableMapInfo(binLogEventsContext);
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            case BinlogEventType.binlog_event_update_rows_event_v1: {
                RowsEvent rowsEvent = new UpdateRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                rowsEvent.setTableMapInfo(binLogEventsContext);
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            case BinlogEventType.binlog_event_delete_rows_event_v1: {
                RowsEvent rowsEvent = new DeleteRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                rowsEvent.setTableMapInfo(binLogEventsContext);
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }


            //
            case BinlogEventType.binlog_event_incident_event: {
                IncidentEvent incidentEvent = new IncidentEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return incidentEvent;
            }
            case BinlogEventType.binlog_event_heartbeat_log_event: {
                HeartbeatEvent heartbeatEvent = new HeartbeatEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return heartbeatEvent;
            }
            case BinlogEventType.binlog_event_ignorable_log_event: {
                IgnorableEvent ignorableEvent = new IgnorableEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return ignorableEvent;
            }
            case BinlogEventType.binlog_event_rows_query_log_event: {
                RowsQueryEvent rowsQueryEvent = new RowsQueryEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return rowsQueryEvent;
            }
            case BinlogEventType.binlog_event_write_rows_event_v2: {
                RowsEvent writeRowsEvent = new WriteRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                writeRowsEvent.setTableMapInfo(binLogEventsContext);
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return writeRowsEvent;
            }
            //
            case BinlogEventType.binlog_event_update_rows_event_v2: {
                RowsEvent rowsEvent = new UpdateRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                rowsEvent.setTableMapInfo(binLogEventsContext);
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            case BinlogEventType.binlog_event_delete_rows_event_v2: {
                RowsEvent rowsEvent = new DeleteRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                rowsEvent.setTableMapInfo(binLogEventsContext);
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            case BinlogEventType.binlog_event_gtid_log_event:
            case BinlogEventType.binlog_event_anonymous_gtid_log_event: {
                GtidEvent newGtidEvent = new GtidEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                GtidSet gtidSet = binLogEventsContext.getGtidSet();
                if (gtidSet != null) {
                    gtidSet.update(newGtidEvent.getGtidString());
                    // update latest gtid
                    header.putGtid(gtidSet, newGtidEvent);
                }
                // update current gtid binlog_event decode context
                binLogEventsContext.setGtidEvent(newGtidEvent);
                return newGtidEvent;
            }
            case BinlogEventType.binlog_event_previous_gtids_log_event: {
                PreviousGtidsEvent previousGtidsEvent = new PreviousGtidsEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return previousGtidsEvent;
            }
            //


            case BinlogEventType.binlog_event_transaction_context_event: {
                TransactionContextEvent transactionContextEvent = new TransactionContextEvent(header, buffer,
                        formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return transactionContextEvent;
            }
            case BinlogEventType.binlog_event_view_change_event: {
                ViewChangeEvent viewChangeEvent = new ViewChangeEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return viewChangeEvent;
            }
            case BinlogEventType.binlog_event_xa_prepare_log_event: {
                XaPrepareEvent xaPrepareEvent = new XaPrepareEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return xaPrepareEvent;
            }
            case BinlogEventType.binlog_event_partial_update_rows_event_v2: {
                RowsEvent rowsEvent = new UpdateRowsEvent(header, buffer, formatDescriptionEvent, true);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                rowsEvent.setTableMapInfo(binLogEventsContext);
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            //
            case BinlogEventType.binlog_event_annotate_rows_event: {
                MariaAnnotateRowsEvent mariaAnnotateRowsEvent = new MariaAnnotateRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                header.putGtid(binLogEventsContext.getGtidSet(), gtidEvent);
                return mariaAnnotateRowsEvent;
            }
            //
            case BinlogEventType.binlog_event_binlog_checkpoint_event: {
                MariaBinlogCheckPointEvent mariaBinlogCheckPointEvent = new MariaBinlogCheckPointEvent(header, buffer,
                        formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return mariaBinlogCheckPointEvent;
            }
            case BinlogEventType.binlog_event_gtid_event: {
                MariaGtidEvent mariaGtidEvent = new MariaGtidEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return mariaGtidEvent;
            }
            case BinlogEventType.binlog_event_gtid_list_event: {
                MariaGtidListEvent mariaGtidListEvent = new MariaGtidListEvent(header, buffer, formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return mariaGtidListEvent;
            }
            case BinlogEventType.binlog_event_start_encryption_event: {
                MariaStartEncryptionEvent mariaStartEncryptionEvent = new MariaStartEncryptionEvent(header, buffer,
                        formatDescriptionEvent);
                /* updating readedIndex in context */
                binlogEventPosition.position = header.getNextBinLogEventPosition();
                return mariaStartEncryptionEvent;
            }
            //
            default:
                /*
                 * Create an object of Ignorable_log_event for unrecognized sub-class. So that
                 * SLAVE SQL THREAD will only update the readedIndex and continue.
                 */
                if ((buffer.getLittleEndian16UnsignedInt(BinLogEvent.flags_offset)
                        & BinLogEvent.log_event_ignorable_f) > 0) {
                    IgnorableEvent ignorableEvent = new IgnorableEvent(header, buffer, formatDescriptionEvent);
                    /* updating readedIndex in context */
                    binlogEventPosition.position = header.getNextBinLogEventPosition();
                    return ignorableEvent;
                } else {
                    if (log.isWarnEnabled()) {
                        log.warn("Skipping unrecognized binlog_event binlog_event " + BinlogEventType.getTypeName(header.getEventType())
                                + " from: " + binLogEventsContext.getBinlogEventPosition());
                    }
                }
        }
        /* updating readedIndex in context */
        binlogEventPosition.position = header.getNextBinLogEventPosition();
        /* Unknown or unsupported log binlog_event */
        return new UnknownEvent(header);
    }

    //
    public BinLogEvent decode(Buffer buffer, BinLogEventsContext binLogEventsContext) throws IOException {
        final int limit = buffer.limit();

        if (limit >= FormatDescriptionEvent.log_event_header_length) {
            Header header = new Header(buffer, binLogEventsContext.getFormatDescriptionEvent());

            final int eventLength = header.getEventLength();
            if (limit >= eventLength) {
                BinLogEvent binLogEvent;
                /**
                 * 有限事件处理
                 */
                if (handleEventTypeSet.get(header.getEventType())) {
                    buffer.newLimit(eventLength);
                    try {
                        binLogEvent = decode(buffer, header, binLogEventsContext);
                    } catch (IOException e) {
                        if (log.isWarnEnabled()) {
                            log.warn("Decoding " + BinlogEventType.getTypeName(header.getEventType()) + " failed from: "
                                    + binLogEventsContext.getBinlogEventPosition(), e);
                        }
                        throw e;
                    } finally {
                        buffer.newLimit(limit); /* Restore newLimit */
                    }
                } else {
                    /* Ignore unsupported binary-log. */
                    binLogEvent = new UnknownEvent(header);
                }

                if (binLogEvent != null) {
                    // set binlogEventFileName
                    binLogEvent.getHeader().setLogFileName(binLogEventsContext.getBinlogEventPosition().getFileName());
                    binLogEvent.setSemiValue(buffer.semiValue);
                }

                /* consume this binary-log. */
                buffer.consume(eventLength);//如果怎样成功或者失败都需要进行消费
                return binLogEvent;
            }
        }

        /* Rewind bytes's readedIndex decode 0. */
        buffer.resetReadedIndex();
        return null;
    }

    public final void handle(final int flagIndex) {
        handleEventTypeSet.set(flagIndex);
    }

    public final void handle(final int fromIndex, final int toIndex) {
        handleEventTypeSet.set(fromIndex, toIndex);
    }


}
